package com.zhku.analysis.service;
/*
 *  @author : 钱伟健 gonefuture@qq.com
 *  @version    : 2018/4/7 21:20.
 *  说明：
 */

import com.zhku.analysis.enyity.UserEveryday;
import com.zhku.analysis.enyity.UserHourly;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.codehaus.jackson.map.util.JSONPObject;
import org.codehaus.jettison.json.JSONArray;
import org.codehaus.jettison.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;

/**
 * <pre> <pre>
 */
@Service
public class ResultService {

    protected static Logger logger= LoggerFactory.getLogger(ResultService.class);

    @Autowired
    private HbaseTemplate hbaseTemplate;

    private UserHourly hourly = new UserHourly();

    private UserEveryday everyday = new UserEveryday();




    public UserEveryday  getEveryday(String rowKey) {
        return  hbaseTemplate.get("everyday",rowKey,
                (result, rowNum) ->{
                    System.out.println("========== rowNum"+rowNum);
                    everyday.setPackagename( Bytes.toString(result.getValue("info".getBytes(), "userId".getBytes())));
                    everyday.setPackagename( Bytes.toString(result.getValue("info".getBytes(), "packagename".getBytes())));
                    everyday.setActivetime( Bytes.toString(result.getValue("info".getBytes(), "activetime".getBytes())));
                    return everyday;
                });
    }


    public List<UserEveryday> getStartRowAndEndRow(final String tableName, final String startKey, final String stopKey, String pageSize) {
        final int ps;
        if (pageSize != null) {
            ps = Integer.parseInt(pageSize);
        } else {
            ps = 50;
        }
        List<UserEveryday> list = new ArrayList<>();
        Scan scan = new Scan();

        scan.setStartRow((startKey+"*").getBytes());// 开始的key
        scan.setStopRow((stopKey+"*").getBytes());// 结束的key
        return hbaseTemplate.execute(tableName, (table) -> {
            int rowNum = 0;
            ResultScanner rscanner = table.getScanner(scan);
            for(Result result : rscanner){
                if (rowNum == ps) break;
                //logger.info("==============" + result);
                UserEveryday userEveryday = new UserEveryday();
                userEveryday.setUserId(new String(result.getValue("info".getBytes(),"userId".getBytes())));
                userEveryday.setPackagename(new String(result.getValue("info".getBytes(),"packagename".getBytes())));
                userEveryday.setActivetime(new String(result.getValue("info".getBytes(),"activetime".getBytes())));
                list.add(userEveryday);
                rowNum ++;
            }
            rscanner.close();
            table.close();
            return list;
            }
        );
    }


    public List<JSONObject>  getJson(String tableName, String startKey, String stopKey, String pageSize) {
        final int ps;
        if (pageSize != null) {
            ps = Integer.parseInt(pageSize);
        } else {
            ps = 50;
        }
        Scan scan = new Scan();
        List<JSONObject>  list = new ArrayList<>();
        scan.setStartRow((startKey+"*").getBytes());// 开始的key
        scan.setStopRow((stopKey+"*").getBytes());// 结束的key
        JSONArray array = new JSONArray();
        return hbaseTemplate.execute(tableName, (table) -> {
            int rowNum = 0;
            ResultScanner rscanner = table.getScanner(scan);
            for(Result result : rscanner){
                if (rowNum == ps) break;
                // 创建json对象就是一个{name:wp}
                JSONObject mapOfConlValues = new JSONObject();
                for (Cell cell: result.rawCells()) {
                    mapOfConlValues.put(Bytes.toString(cell.getQualifier()),
                            Bytes.toString(cell.getValue()));
                    //System.out.println("QualifierArray=================== "+Bytes.toString(cell.getQualifierArray()));
                    //System.out.println("getValueArray=================== "+Bytes.toString(cell.getValueArray()));
                }
                rowNum ++;
                list.add(mapOfConlValues);
            }
            // 关闭查询
            rscanner.close();
            table.close();
            //System.out.print("=============  list  "+list);
            return list;
        });

    }
}

